package com.wolf.etl.core.transform;

import com.alibaba.fastjson.JSONObject;
import com.wolf.etl.core.WriteBackManager;
import com.wolf.etl.core.dialect.AbstractDialect;
import com.wolf.etl.core.handler.HandlerChain;
import com.wolf.etl.core.rule.check.AbstractCheckRule;
import com.wolf.etl.core.rule.convert.AbstractConvertRule;
import com.wolf.etl.enumeration.APPLICABLE_SCOPE;
import com.wolf.etl.enumeration.CHECK_RULE_TYPE;
import com.wolf.etl.enumeration.CONVERT_RULE_TYPE;
import com.wolf.etl.model.*;
import com.wolf.etl.vo.BuildSqlParam;
import com.wolf.etl.vo.ColumnDesc;
import com.wolf.exception.ETL_ERROR_MSG;
import com.wolf.util.StringUtils;
import com.wolf.vo.BaseResult;
import com.wolf.vo.KeyValuePair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author sdyang
 * @create 2019-06-13 10:24
 **/
@Service("mapTransform")
public class MapTransform extends AbstractTransform {

    private static Logger logger = LoggerFactory.getLogger(MapTransform.class);

    private static final String logger_prefix = "wolf-etl#com.wolf.etl.core.transform#MapTransform";

    @Override
    public String getCountSql(){

        BuildSqlParam param= initParam();

        AbstractDialect srcDialect = AbstractDialect.getInstance(manager, manager.getDataSourceDialet(srcDataSourceId));
        srcDialect.setDatasourceId(srcDataSourceId);

        String sql = srcDialect.buildCount(param);
        return sql;
    }

    /**
     *
     * @return
     */
    private BuildSqlParam initParam(){

        Map<Long, String> srcFields = new HashMap<>();
        List<String> tarFields = new ArrayList<String>();
        for (EtlTransformDefineModel vo : transformDefineModels) {
            srcFields.put(vo.getSrc_table_id(), vo.getSrc_column_name());
            tarFields.add(vo.getTarget_column_name());
        }
        EtlTableModel srcTable = manager.getTableModel();

        Map<Long, EtlTableModel> relMap = new HashMap<Long, EtlTableModel>();
        relMap.put(srcTable.getId(), srcTable);
        List<EtlTableRelModel> rels = manager.findEtlTableSlaveRelModels(srcTable.getId());
        for (EtlTableRelModel rel : rels) {
            if (!relMap.containsKey(rel.getMaster_table_id())) {
                EtlTableModel table = manager.findEtlTableModel(rel.getMaster_table_id());
                relMap.put(rel.getMaster_table_id(), table);
            }
            if (!relMap.containsKey(rel.getSlave_table_id())) {
                EtlTableModel table = manager.findEtlTableModel(rel.getSlave_table_id());
                relMap.put(rel.getSlave_table_id(), table);
            }

        }

        BuildSqlParam param = new BuildSqlParam();
        param.setSrcTable(srcTable);
        param.setFetch_max_count(manager.getEtlDefineModel().getFetch_max_count());
        param.setOffset(manager.getNextPage()*manager.getEtlDefineModel().getFetch_max_count());
        param.setFilters(filters);
        param.setOrderbys(orderbys);
        param.setSrcFields(srcFields);
        param.setTransformDefineModelList(transformDefineModels);
        param.setRels(rels);
        param.setRelMap(relMap);
        param.setRuleTypeMap(manager.findRuleTypeMap());
        param.setTransformRules(manager.findTransRules());
        param.setTableRelRules(manager.findTableRelRules());

        param.init();

        return param;
    }

    @Override
    public BaseResult doTransform() {
        BaseResult result = BaseResult.getSuccess();

        try {
            AbstractDialect srcDialect = AbstractDialect.getInstance(manager, manager.getDataSourceDialet(srcDataSourceId));
            srcDialect.setDatasourceId(srcDataSourceId);

            EtlTableModel srcTable = manager.getTableModel();

            BuildSqlParam param = initParam();

            String sql = srcDialect.buildSelect(param);
            this.sql = sql;

            // 字段回写原表
            srcDialect.addSql(new KeyValuePair(srcTable.getId() + "", sql));

            srcData = manager.queryResults(sql);

            // 数据处理（责任链）
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("data", srcData);
            HandlerChain handlerChain = new HandlerChain(manager);
            jsonObject = handlerChain.done(jsonObject);
            srcData = (List<Map<String, ColumnDesc>>) jsonObject.get("data");

            Map<String, EtlTransformDefineModel> transMap = new HashMap<>();
            for (EtlTransformDefineModel vo : transformDefineModels) {
                String src_column = StringUtils.isNotEmpty(vo.getSrc_column_alias()) ? vo.getSrc_column_alias() : vo.getSrc_column_name();
                transMap.put(src_column.toLowerCase(), vo);
            }

            EtlTableModel targerTable = manager.findEtlTableModel(transformDefineModels.get(0).getTarget_table_id());

            AbstractDialect targetDialect = AbstractDialect.getInstance(manager, manager.getDataSourceDialet(targetDataSourceId));

            List<Object> rewritebackIds = new ArrayList<Object>();
            boolean flag = true;
            for (Map<String, ColumnDesc> record : srcData) {
                Map<String, ColumnDesc> fields = new HashMap<String, ColumnDesc>();
                for (Map.Entry<String, ColumnDesc> column : record.entrySet()) {

                    //回写源表主键的字段值
                    if (column.getKey().equals(srcTable.getPrimary_key())) {
                        rewritebackIds.add(column.getValue().getObj());
                    }

                    EtlTransformDefineModel vo = transMap.get(column.getKey().toLowerCase());

                    if (vo == null) {
                        continue;
                    }

                    String value = column.getValue().getObj() == null ? null : column.getValue().getObj().toString();

                    // 校验规则
                    List<EtlTransformRuleModel> rules = param.getCheckRuleMap().get(vo.getId());

                    if (rules != null) {
                        for (EtlTransformRuleModel rule : rules) {
                            EtlTransformRuleTypeModel typeModel = param.getRuleTypeMap().get(rule.getType_id());

                            AbstractCheckRule checkRule = AbstractCheckRule.getInstance(CHECK_RULE_TYPE.valueOf(typeModel.getCode()), manager);
                            BaseResult checkResult = checkRule.check(rule.getTransform_cxt(), value);
                            if (checkResult.isFailure()) {
                                manager.error(vo.getEtl_define_id(), vo.getSrc_table_id(), vo.getSrc_column_name(), null, ETL_ERROR_MSG.ERROR_TRANSFORM_RULE_001.toString(), checkResult.getMessage());
                                flag = false;
                                break;
//                            WolfException.report(ETL_ERROR_MSG.ERROR_TRANSFORM_RULE_001,checkResult.getMessage());
                            } else {
                                flag = true;
                            }

                        }
                    }

                    // 转换规则
                    rules = param.getConvertRuleMap().get(vo.getId());
                    if (rules != null) {
                        for (EtlTransformRuleModel rule : rules) {

                            EtlTransformRuleTypeModel typeModel = param.getRuleTypeMap().get(rule.getType_id());

                            if(!typeModel.getApplicable_scope().equalsIgnoreCase(APPLICABLE_SCOPE.Target.toString())){
                                continue;
                            }

                            AbstractConvertRule convertRule = AbstractConvertRule.getInstance(CONVERT_RULE_TYPE.valueOf(typeModel.getCode()), manager);
                            AbstractConvertRule.RetInfo retInfo = convertRule.convert(rule.getTransform_cxt(), value, null);
                            value = retInfo.getTargetInfo();
                            column.getValue().setObj(value);
                            column.getValue().setFormula(true);
                            flag = true;
                        }
                    }

                    if (flag) {
                        fields.put(vo.getTarget_column_name(), column.getValue());
                    }

                }
                String insertSql = targetDialect.buildInsert(targerTable, fields);

                System.out.println(insertSql);
                manager.appendTargetSql(insertSql);

            }

            // 回写源表字段信息
            Map<String, String> rewriteBackSqlMap = WriteBackManager.getInstance(manager).writeBack(param, rewritebackIds);
            for (Map.Entry<String, String> entry : rewriteBackSqlMap.entrySet()) {
                System.out.println(entry.getValue());
                manager.appendWriteBackSql(entry.getValue());
            }
        }catch (Exception e){
            e.printStackTrace();
        }

        return result;
    }
}
